package com.shujia.flink.window;

import com.mysql.jdbc.log.Log;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Int;

/**
 * @author shujia
 */
public class Demo8WindowApply {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> lines = env.socketTextStream("master", 8888);

        DataStream<Tuple2<String, Integer>> kv = lines.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyBy = kv.keyBy(k -> k.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windows = keyBy
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        //匿名内部类
        DataStream<CountResult> countResults = windows
                .apply(new WindowFunction<Tuple2<String, Integer>, CountResult, String, TimeWindow>() {
                    /**
                     * apply: 每一个key每个窗口执行一次
                     * @param word :单词
                     * @param window : 窗口对象,可以获取窗口的开始和结束时间
                     * @param input : 一个窗口内哦的数据
                     * @param out :将数据发生到下游
                     */
                    @Override
                    public void apply(String word, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<CountResult> out) throws Exception {

                        //窗口的开始和结束时间
                        long start = window.getStart();
                        long end = window.getEnd();

                        int count = 0;
                        //循环窗口内的数据,计算单词的数量
                        for (Tuple2<String, Integer> kv : input) {
                            count++;
                        }

                        //将数据发生到下游
                        CountResult countResult = new CountResult(word, count, start, end);

                        out.collect(countResult);
                    }
                });


        //lambda表达式
        DataStream<CountResult> countResults1 = windows
                .apply((word, window, input, out) -> {
                    //窗口的开始和结束时间
                    long start = window.getStart();
                    long end = window.getEnd();

                    int count = 0;
                    //循环窗口内的数据,计算单词的数量
                    for (Tuple2<String, Integer> kv1 : input) {
                        count++;
                    }

                    //将数据发生到下游
                    CountResult countResult = new CountResult(word, count, start, end);

                    out.collect(countResult);
                }, TypeInformation.of(CountResult.class));

        countResults1.print();

        env.execute();


    }

    @Data
    @AllArgsConstructor
    static class CountResult {
        private String word;
        private int count;
        private long start;
        private long end;
    }
}
